package vn.com.epi.recommender;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.data.table.ArticleTable;
import vn.com.epi.data.table.NeighbourTable;
import vn.com.epi.data.table.RecommendationTable;
import vn.com.epi.data.table.TempUserTable;
import vn.com.epi.data.table.UserTable;
import vn.com.epi.utils.StringUtils;
import vn.com.epi.utils.TableUtils;

import com.google.common.base.Strings;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;

public class BaomoiRecommender extends Configured implements Tool {

	private static final int MAX_GROUP_SIZE = 1000;
	
	private Map<Integer, Integer> category;
	private Map<String, String> nearest;
	private Map<String, String> oldDistribution;
	private Multimap<String, Integer> history;
	private int[] categoryCode = {45,47,50,52,53,54,55,58,59,71,72,76,79,82,84,87,88,104,107,108,119,121,122,126,127,132,136,137,138,139,141,144,145,146,147,148,150,152};
	
	private static class Pair implements Comparable<Pair> {
		
		Object key;
		int value;
		
		public Pair(Object key_, int value_) {
			key = key_;
			value = value_;
		}

		@Override
		public int compareTo(Pair o) {
			if (value > o.value)
				return -1;
			else
				if (value < o.value)
					return 1;
			return 0;
		}
	}

	public BaomoiRecommender() {
		category = Maps.newHashMap();
		nearest = Maps.newHashMap();
		oldDistribution = Maps.newHashMap();
		history = ArrayListMultimap.create();
	}
	
	private int getcategoryId(int categoryID) {
		int first = 0;
		int last = categoryCode.length - 1;
		while (first <= last) {
			int mid = (first + last)/2;
			if (categoryCode[mid] == categoryID)
				return mid + 1;
			else
				if (categoryCode[mid] < categoryID)
					first = mid + 1;					
				else
					last = mid - 1;					
		}
		int index = -1;
		return index;
	}		

	private void loadCategory() throws IOException {
        HTable table = new HTable(ArticleTable.NAME);
		try {
			Scan scan = new Scan();
			scan.addColumn(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID);
			ResultScanner rs = table.getScanner(scan);
			for (Result r : rs) {
				int articleId = Bytes.toInt(r.getRow());
				int categoryId = Bytes.toInt(r.getValue(ArticleTable.FAM_MAIN, ArticleTable.COL_CATEGORY_ID));
				categoryId = getcategoryId(categoryId);
				if (categoryId < 0)
					continue;
				category.put(articleId, categoryId);
			}
		} finally {
			table.close();
		}
	}
	
	private void getNearestNeighbour() throws IOException {
		HTable table = new HTable(NeighbourTable.NAME);
		try {
            Scan scan = new Scan();
            scan.addColumn(NeighbourTable.FAM_MAIN, NeighbourTable.COL_NEAREST);        
            ResultScanner rs = table.getScanner(scan);
            for (Result r: rs) {
            	String userId = Bytes.toString(r.getRow());
            	String neighbours = NeighbourTable.getNearest(r);
            	if (!Strings.isNullOrEmpty(neighbours)) {
            		nearest.put(userId, neighbours);
            	}
            }
		} finally {
			table.close();			
		}		            
	}
	
	private void readHistoryFromHbase() throws IOException {				
		HTable table = new HTable(TempUserTable.NAME);
		try {
            Scan scan = new Scan();
            scan.addColumn(TempUserTable.FAM_MAIN, TempUserTable.COL_VISITED_ARTICLES);
            scan.addColumn(TempUserTable.FAM_MAIN, TempUserTable.COL_DISTRIBUTION);
            ResultScanner rs = table.getScanner(scan);
            for (Result r: rs) {
            	String userId = Bytes.toString(r.getRow());
            	String line = TempUserTable.getVisitedArticles(r);
            	if (Strings.isNullOrEmpty(line))
            		continue;
            	String[] articles = line.split(",");
            	for (int i = 0; i < articles.length; i++) {
            		int articleId = Integer.parseInt(articles[i]);
            		history.put(userId, articleId);
				}
            	line = TempUserTable.getDistribution(r);
            	if (!Strings.isNullOrEmpty(line))
            		oldDistribution.put(userId, line);            		
            }

		} finally {
			table.close();			
		}		
	}
	
	@SuppressWarnings("unused")
	private String[] recommend(String userA, String userB, int k) {
		Map<Integer, Boolean> map = Maps.newHashMap();
		StringBuilder sb;
		sb = new StringBuilder();
		if (history.containsKey(userA)) 
			for (int i: history.get(userA))
				map.put(i, true);
		if (history.containsKey(userB)) 
			for (int i: history.get(userB))
				if (!map.containsKey(i))
					sb.append(i).append(",");
		if (sb.length() > 0) {
			sb.deleteCharAt(sb.length() - 1);
			return sb.toString().split(",");
		} else
			return new String[0];
	}

	
	private String[] recommend(String userA, String userB) {
		Map<Integer, Boolean> map = Maps.newHashMap();
		StringBuilder sb;
		sb = new StringBuilder();
		if (history.containsKey(userA)) 
			for (int i: history.get(userA))
				map.put(i, true);
		if (history.containsKey(userB)) 
			for (int i: history.get(userB))
				if (!map.containsKey(i))
					sb.append(i).append(",");
		if (sb.length() > 0) {
			sb.deleteCharAt(sb.length() - 1);
			return sb.toString().split(",");
		} else
			return new String[0];
	}
	
	private Map<Integer, Integer> getDistribution(String userId) {
		Map<Integer, Integer> map = Maps.newHashMap();
		if (oldDistribution.containsKey(userId)) {
			String line = oldDistribution.get(userId);
			String[] chunks = line.split(" ");
			for (int i = 0; i < chunks.length; i++) {
				String[] val = chunks[i].split(",");
				int categoryId = Integer.parseInt(val[0]);
				int freq = Integer.parseInt(val[1]);
				map.put(categoryId, freq);
			}
		}		
		for (int articleid : history.get(userId))
			if (category.containsKey(articleid)) {
				int cat = category.get(articleid);
				if (map.containsKey(cat)) {
					int num = map.get(cat);
					num++;
					map.put(cat, num);
				} else
					map.put(cat, 1);
			}
		return map;
	}
	
	private String recommendations(String user, int k) {
		Map<Integer, Integer> map = Maps.newHashMap();
		Multimap<Integer, Integer> articleByCategory = HashMultimap.create();
		List<Integer> list;
		StringBuilder sb;
		String s;
		Map<Integer, Integer> distribution = getDistribution(user);
		
		if (!nearest.containsKey(user))
			return null;
		else {
			s = nearest.get(user);
			String[] nearUsers = s.split(",");
			for (int i = 0; i < nearUsers.length; i++) {
				String[] items = recommend(user, nearUsers[i]);
				for (int j = 0; j < items.length; j++) {
					int item = Integer.parseInt(items[j]);
					if (map.containsKey(item)) {
						int num = map.get(item);
						num++;
						map.put(item, num);
					} else
						map.put(item, 1);
				}
			}
								
			List<Pair> topRead = Lists.newArrayList();
			for (int cat: distribution.keySet()) {
				int freq = distribution.get(cat);
				topRead.add(new Pair(cat, freq));
			}
			Collections.sort(topRead);
			
			for (Map.Entry<Integer, Integer> entry: map.entrySet()) {
				int article = entry.getKey();
				int cat = 0;
				if (category.containsKey(article))  
					cat = category.get(article);
				if (cat != 0) {
					articleByCategory.put(cat, article);
				}
			}

			Map<Integer, Boolean> choose = Maps.newHashMap();
			list = new ArrayList<Integer>();
			int numCategory = 0;
			int num = 7;
			for (Pair entry: topRead) {
				int cat = (Integer) entry.key;
				int kArticle = Math.min(num, articleByCategory.get(cat).size());
				@SuppressWarnings({ "unchecked", "rawtypes" })
				List<Integer> arts = new ArrayList(articleByCategory.get(cat));
				Collections.shuffle(arts);
				int counter = 0;
				for (int art : arts) {
					choose.put(art, true);
					list.add(art);
					counter++;
					if (counter == kArticle)
						break;
				}
				num--;
				numCategory++;
				if (numCategory == 5) break;
			}
			
			List<Integer> set = new ArrayList<Integer>(map.keySet());
			Collections.shuffle(set);
			if (list.size() < k) {
				int counter = k - list.size();
				for (int article: set) 
					if (!choose.containsKey(article)) {
						choose.put(article, true);
						list.add(article);
						counter--;
						if (counter == 0)
							break;
					}
			}

			Collections.shuffle(list);
			sb = new StringBuilder();
			int count = 0;
			for (int art: list) {
				sb.append(art).append(",");
				count++;
				if (count == k)
					break;
			}
			if (sb.length() > 0)
				sb.deleteCharAt(sb.length() - 1);
			return sb.toString();
		}
	}
	
	private List<String> getTopArticles(Map<String, Integer> frequency, int howMany) {
		List<Pair> list;
		List<String> top;
		int k = Math.min(frequency.size(), howMany);
		list = Lists.newArrayList();
		top = Lists.newArrayList();
		if (k == 0)
			return top;
		for (Map.Entry<String, Integer> entry: frequency.entrySet()) {
			list.add(new Pair(entry.getKey(), entry.getValue()));
		}
		Collections.sort(list);
		int count = 0;
		for (Pair entry: list) {
			top.add((String)entry.key);
			count++;
			if (count == k)
				break;
		}
		return top;
	}

	@Override
	public int run(String[] arg0) throws Exception {
		int HOW_MANY = 10;
		Map<String, Integer> frequency;
		Map<String, String> suggestion;
		loadCategory();
		readHistoryFromHbase();
		getNearestNeighbour();
		frequency = Maps.newHashMap();
		suggestion = Maps.newHashMap();
		for (String userId: history.keySet()) {
			String res = recommendations(userId, HOW_MANY);			
			if (!Strings.isNullOrEmpty(res)) {
				suggestion.put(userId, res);
				String[] chunks = res.split(",");
				for (int i = 0; i < chunks.length; i++) {
					if (frequency.containsKey(chunks[i])) {
						int num = frequency.get(chunks[i]);
						num++;
						frequency.put(chunks[i], num);
					} else
						frequency.put(chunks[i], 1);
				}
			}
		}		
		TableUtils.truncateTable(RecommendationTable.NAME_BYTES);
		List<String> top = getTopArticles(frequency, 30);
		HTable table = new HTable(RecommendationTable.NAME);
		List<Put> puts = Lists.newArrayList();
		try {
			for (String userId: suggestion.keySet()) {
				String res = suggestion.get(userId);
				String[] chunks = res.split(",");
				StringBuilder sb = new StringBuilder();
				sb.append(res);
				if (chunks.length < HOW_MANY) {
					sb.append(",");
					Collections.shuffle(top);
					int k = HOW_MANY - chunks.length;
					for (String articleId: top) {
						sb.append(articleId).append(",");
						k--;						
						if (k == 0)
							break;
					}
					StringUtils.deleteLastChar(sb);
				}
				byte[] row = Bytes.toBytes(userId);
				byte[] value = Bytes.toBytes(sb.toString());
				Put put = new Put(row);
				put.add(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS, value);
				puts.add(put);
				if (puts.size() == MAX_GROUP_SIZE) {
					table.put(puts);
					puts = Lists.newArrayList();
				}
			}
			StringBuilder sb = new StringBuilder();
			for (String articleId: top)
				sb.append(articleId).append(",");
			StringUtils.deleteLastChar(sb);
			byte[] row = Bytes.toBytes(UserTable.COLD_START);
			Put put = new Put(row);
			byte[] value = Bytes.toBytes(sb.toString());
			put.add(RecommendationTable.FAM_MAIN, RecommendationTable.COL_RECOMMENDATIONS, value);
			puts.add(put);					
			table.put(puts);
		} finally {
			table.close();
		}
		return 0;
	}
	
	public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        ToolRunner.run(new BaomoiRecommender(), args);
        long endTime = System.currentTimeMillis();
        System.out.println("total time used for recommend: " + (endTime - startTime) + " ms.");
		
	}

 }
